#include "app_pool.h"
#include "mqueue.h"
#include "pthread.h"
#include "log.h"
#include "errno.h"
#include "stdlib.h"

static mqd_t mqd;
static pthread_t threads[THREAD_NUM];

/**
 * 每个线程 要持续做的事情： 1 循环从队列中获取任务，2 执行任务
 */
static void *app_pool_worker_func(void *arg)
{   
    log_info("app_pool thread start");
    while(1)
    {
        Task task;
        //1 从队列中获取任务
        mq_receive(mqd, (char *)&task, sizeof(Task), 0); 
        //2 执行任务
        task.task_func(task.arg);
    }
    return NULL;
}

/**
 * 初始化线程 池
 * 1  创建队列 
 * 2  创建n线程数组
 */
int app_pool_init()
{
    //0 清理久的队列文件
    mq_unlink(MQ_NAME);
// * 1  创建队列 
    struct mq_attr attr;
    attr.mq_maxmsg = MQ_MAXSIZE;
    attr.mq_msgsize = sizeof(Task);
    mqd= mq_open(MQ_NAME,O_CREAT|O_RDWR, 0666, &attr);
    if(mqd == -1)
    {
        perror("mq_open failed");
        return -1;
    }

    //2 初始化线程数组
    for(int i=0;i<THREAD_NUM;i++)
    {
       int ret= pthread_create(&threads[i],NULL,app_pool_worker_func,NULL);
        if(ret != 0)
        {
            perror("pthread_create failed");
            return -1;
        }
    }
    return 0;

}

/**
 *  向线程池添加任务
 *   1   把函数和参数封装为task结构体
 *   2   把task结构体放入队列中
 */
int app_pool_add_task(void *(*task_func)(void *arg), void *arg)
{   //1   把函数和参数封装为task结构体
    Task *task = (Task *)malloc(sizeof(Task));
    task->task_func = task_func;
    task->arg = arg;
    //2   把task结构体放入队列中
    int ret = mq_send(mqd, (char *)task, sizeof(Task),0);
    if(ret == -1)
    {
        perror("mq_send failed");
        return -1;
    }
    return 0;

}

/**
 *  销毁线程池
 * 1  结束所有的线程 
 * 2  销毁队列  删除队列文件
 */
int app_pool_destroy()
{
    for (size_t i = 0; i < THREAD_NUM; i++)
    {
         pthread_cancel(threads[i]);
         pthread_join(threads[i],NULL);
    }

    mq_close(mqd);
    mq_unlink(MQ_NAME);
    return 0;
    
}